package com.xinxin.blockchain.executor;

import com.google.common.collect.Lists;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.List;
import java.util.concurrent.Callable;
import java.util.function.Supplier;

/**
 * @author ys
 * @date 2020/6/22
 */
@Slf4j
@Component
public class LockExecutor {

    private static RedissonClient redissonClient;

    @Autowired
    public void setRedissonClient(RedissonClient redissonClient) {
        LockExecutor.redissonClient = redissonClient;
    }

    public static void runWithLock(Runnable runnable, String... keys) {
        runWithLock(Lists.newArrayList(keys), runnable);
    }

    public static void runWithLock(List<String> keys, Runnable runnable) {
        runWithLock(defaultKeyGenerator(keys), runnable);
    }

    public static void runWithLock(Supplier<String> keyGenerator, Runnable runnable) {
        runWithLock(keyGenerator.get(), runnable);
    }

    public static void runWithLock(String lockKey, Runnable runnable) {
        RLock lock = redissonClient.getLock(lockKey);

        lock(lock);
        try {
            runnable.run();
        } finally {
            unlock(lock);
        }
    }

    public static <V> V runWithLock(Callable<V> callable, String... keys) throws Exception {
        return runWithLock(Lists.newArrayList(keys), callable);
    }

    public static <V> V runWithLock(List<String> keys, Callable<V> callable) throws Exception {
        return runWithLock(defaultKeyGenerator(keys), callable);
    }

    public static <V> V runWithLock(Supplier<String> keyGenerator, Callable<V> callable) throws Exception {
        return runWithLock(keyGenerator.get(), callable);
    }

    public static <V> V runWithLock(String lockKey, Callable<V> callable) throws Exception {
        RLock lock = redissonClient.getLock(lockKey);

        lock(lock);
        try {
            return callable.call();
        } finally {
            unlock(lock);
        }
    }

    public static void tryRunWithLock(Runnable runnable, String... keys) {
        tryRunWithLock(Lists.newArrayList(keys), runnable);
    }

    public static void tryRunWithLock(List<String> keys, Runnable runnable) {
        tryRunWithLock(defaultKeyGenerator(keys), runnable);
    }

    public static void tryRunWithLock(Supplier<String> keyGenerator, Runnable runnable) {
        tryRunWithLock(keyGenerator.get(), runnable);
    }

    public static void tryRunWithLock(String lockKey, Runnable runnable) {
        RLock lock = redissonClient.getLock(lockKey);

        boolean acquired = tryLock(lock);
        if (!acquired) {
            return;
        }

        try {
            runnable.run();
        } finally {
            unlock(lock);
        }
    }

    public static <V> V tryRunWithLock(Callable<V> callable, String... keys) throws Exception {
        return tryRunWithLock(Lists.newArrayList(keys), callable);
    }

    public static <V> V tryRunWithLock(List<String> keys, Callable<V> callable) throws Exception {
        return tryRunWithLock(defaultKeyGenerator(keys), callable);
    }

    public static <V> V tryRunWithLock(Supplier<String> keyGenerator, Callable<V> callable) throws Exception {
        return tryRunWithLock(keyGenerator.get(), callable);
    }

    public static <V> V tryRunWithLock(String lockKey, Callable<V> callable) throws Exception {
        RLock lock = redissonClient.getLock(lockKey);

        boolean acquired = tryLock(lock);
        if (!acquired) {
            return null;
        }

        try {
            return callable.call();
        } finally {
            unlock(lock);
        }
    }

    private static Supplier<String> defaultKeyGenerator(List<String> keys) {
        if (keys == null) {
            throw new NullPointerException("key list is null.");
        }

        return () -> keys.stream()
                .filter(StringUtils::isNotBlank)
                .reduce((a, b) -> a + "_" + b)
                .orElseThrow(() -> new IllegalStateException("lock keys empty."));
    }

    private static void lock(RLock lock) {
        log.info("lock key: {}", lock.getName());
        lock.lock();
        log.info("locked: {}", lock.getName());
    }

    private static boolean tryLock(RLock lock) {
        log.info("try lock key: {}", lock.getName());
        boolean acquired = lock.tryLock();

        log.info("try lock {}: {}", lock.getName(), acquired);
        return acquired;
    }

    private static void unlock(RLock lock) {
        lock.unlock();
        log.info("unlock: {}", lock.getName());
    }
}
